package com.cloud.canal.mes;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.cloud.service.CanalDataHandlerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;

@Component
@Slf4j
public class MesCanalClient implements InitializingBean {
    @Autowired
    private CanalDataHandlerService canalDataHandlerService;

    @Value("${canal.server.mes.ip:127.0.0.1}")
    private String SERVER_ADDRESS;

    /**
     * canal server 端口号
     */
    @Value("${canal.server.mes.port:11111}")
    private Integer PORT;

    /**
     * DESTINATION 你配置的监听文件名称 这个是你canal下面的 文件夹的名称。如果你没写名称的话
     */
    @Value("${canal.server.mes.destination:canal-test}")
    private String DESTINATION;

    /**
     * canal 用户名
     */
    @Value("${canal.server.mes.username:canal}")
    private String USERNAME;

    /**
     * canal 密码
     */
    @Value("${canal.server.mes.password:canal}")
    private String PASSWORD;

    @Value("${canal.server.mes.batch-size:1000}")
    private int BATCH_SIZE;

    @Value("${canal.server.mes.subscribe:null}")
    private String filter;

    @Value("${canal.server.mes.sleep-time:10000}")
    private Integer sleepTime;

    public CanalConnector connector = null;

    /**
     * 启动前  创建canal连接
     */
    @Override
    public void afterPropertiesSet() {
        getConnect();
    }

    /**
     * 获取链接
     */
    public CanalConnector getConnect() {

        if (connector == null) {
            // 创建链接
            // 这里填写canal所配置的服务器ip,端口号,destination(在canal.properties文件里)以及服务器账号密码
            connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD);
        }
        // 校验连接状态
//        log.info("canalServer连接状态：" + connector.checkValid());
        return connector;
    }

    /**
     * 订阅表  如.*\\..*
     */
    public void subscribe(String filter) {
        if (!StringUtils.isEmpty(filter)) {

            //订阅数据库表,全部表,server端配置了，则该处无需配置，否则不生效
            connector.subscribe(filter);
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
            connector.rollback();
        }
    }

    public void getMessage() {
        try {
            //打开连接
            connector.connect();
//            connector.subscribe(null);
            subscribe(filter);
            while (true) {
                //每隔10秒拉一次数据
                Thread.sleep(sleepTime);
                //获取指定数量的数据
                Message message = connector.getWithoutAck(BATCH_SIZE);
                try {
                    // 数据处理
                    canalDataHandlerService.handleMessage(message, connector);
                } catch (Exception e) {
                    log.error("该批数据处理失败");
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            // todo 配置预警
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
}